package bigdata.monitor.binlog.utils;

import lombok.extern.slf4j.Slf4j;
import org.sqlite.SQLiteConfig;

import java.sql.*;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 基于 SQLite in-memory 的 JSON 规则表达式求值器。
 * 支持使用 JSON Path（如 $.before.field）在 SQL 条件中进行字段比较。
 * 在执行前自动校验所有 JSON 路径是否存在，避免因路径错误导致规则静默失效。
 */
@Slf4j
public class SqliteJsonRuleEvaluator {

    /**
     * SQLite 内存数据库连接 URL
     */
    private static final String SQLITE_URL = "jdbc:sqlite::memory:";

    /**
     * 每个线程持有一个只读的 SQLite 内存数据库连接
     */
    private static final ThreadLocal<Connection> connHolder = ThreadLocal.withInitial(() -> {
        try {
            SQLiteConfig config = new SQLiteConfig();
            config.setReadOnly(true); // 必须在创建连接前设置只读
            Connection conn = config.createConnection(SQLITE_URL);
            // 注册 REGEXP 函数
            org.sqlite.Function.create(conn, "REGEXP", new org.sqlite.Function() {
                @Override
                protected void xFunc() throws SQLException {
                    String pattern = value_text(0);
                    String value = value_text(1);
                    if (value == null) {
                        result(0);
                        return;
                    }
                    try {
                        result(java.util.regex.Pattern.matches(pattern, value) ? 1 : 0);
                    } catch (Exception e) {
                        result(0);
                    }
                }
            });
            return conn;
        } catch (SQLException e) {
            throw new IllegalStateException("Failed to initialize SQLite in-memory database connection", e);
        }
    });

    /**
     * 用于匹配标准点号风格的 JSON Path，例如：$.a.b, $.before.contact_info
     * 要求：
     * - 以 "$." 开头
     * - 后续由字母、数字、下划线、点号组成（不包含空格、引号、方括号等）
     * - 不允许连续点（..）或结尾为点
     */
    private static final Pattern JSON_PATH_PATTERN = Pattern.compile("\\$\\.[\\w.]+");

    /**
     * 从给定的条件表达式中提取所有符合 $.xxx.yyy 格式的 JSON 路径
     *
     * @param expression 包含 JSON Path 的条件表达式，例如：
     *                   "$.before.contact_info != $.after.contact_info"
     * @return 去重后的路径集合，例如：{"$.before.contact_info", "$.after.contact_info"}
     */
    private static Set<String> extractJsonPaths(String expression) {
        if (expression == null) {
            return Collections.emptySet();
        }
        Set<String> paths = new LinkedHashSet<>();
        Matcher matcher = JSON_PATH_PATTERN.matcher(expression);
        while (matcher.find()) {
            String path = matcher.group();
            // 过滤非法形式：不能以点结尾，不能包含连续点
            if (!path.endsWith(".") && !path.contains("..")) {
                paths.add(path);
            }
        }
        return paths;
    }

    /**
     * 校验 JSON 路径是否存在，采用快速失败策略：只要发现一个路径不存在，立即返回 false。
     *
     * @param jsonPayload JSON 字符串
     * @param paths       要校验的 JSON 路径集合
     * @return true 表示所有路径均存在；false 表示至少第一个缺失路径已记录
     */
    private static boolean validateJsonPaths(String jsonPayload, Set<String> paths) {
        if (paths == null || paths.isEmpty()) {
            return true;
        }

        Connection conn = connHolder.get();

        for (String path : paths) {
            PreparedStatement stmt = null;
            ResultSet rs = null;
            try {
                stmt = conn.prepareStatement("SELECT json_type(?, ?)");
                stmt.setString(1, jsonPayload);
                stmt.setString(2, path);
                rs = stmt.executeQuery();

                if (!rs.next() || rs.getString(1) == null) {
                    log.warn("JSON path does not exist: {}, json preview: {}",
                            path,
                            jsonPayload != null ? jsonPayload.substring(0, Math.min(100, jsonPayload.length())) : "null");
                    return false; // ✅ 快速失败：发现一个缺失就返回
                }
            } catch (SQLException e) {
                log.warn("Error while validating JSON path: {}", path, e);
                return false;
            } finally {
                // 安全关闭资源
                if (rs != null) {
                    try { rs.close(); } catch (SQLException ignored) {}
                }
                if (stmt != null) {
                    try { stmt.close(); } catch (SQLException ignored) {}
                }
            }
        }
        return true;
    }

    /**
     * 执行 JSON 条件表达式求值
     *
     * @param jsonPayload      待评估的 JSON 字符串，必须为合法 JSON
     * @param jsonPathCondition SQL WHERE 子句，允许使用 "data" 作为 JSON 列名，并引用 JSON Path，
     *                         例如：json_extract(data, '$.before.contact_info') != json_extract(data, '$.after.contact_info')
     *                         或更简洁地，由调用方保证表达式中路径已转换为 json_extract 形式。
     *                         ⚠️ 注意：本方法会自动提取 $.xxx 路径做存在性校验，但表达式本身必须是合法 SQLite 表达式。
     * @return 如果条件成立返回 true，否则返回 false；执行出错或路径缺失也返回 false
     */
    public static boolean evaluate(String jsonPayload, String jsonPathCondition) {
        if (jsonPayload == null || jsonPathCondition == null) {
            return false;
        }

        // 1. 从条件表达式中提取所有 $.xxx.yyy 形式的路径
        Set<String> requiredPaths = extractJsonPaths(jsonPathCondition);

        // 2. 校验所有路径在当前 JSON 中是否存在
        if (!validateJsonPaths(jsonPayload, requiredPaths)) {
            // 路径缺失已通过 warn 日志上报，此处不再执行表达式，直接返回 false
            return false;
        }

        // 3. 构造可执行的 SQL：将 jsonPayload 作为虚拟表的 data 列
        String sql = "SELECT 1 FROM (SELECT ? AS data) WHERE " + jsonPathCondition;

        Connection conn = connHolder.get();
        try (PreparedStatement stmt = conn.prepareStatement(sql)) {
            stmt.setString(1, jsonPayload);
            try (ResultSet rs = stmt.executeQuery()) {
                return rs.next(); // 有结果即条件成立
            }
        } catch (SQLException e) {
            log.warn("SQLite evaluation failed. Condition: {}, JSON preview: {}",
                    jsonPathCondition,
                    jsonPayload.length() > 100 ? jsonPayload.substring(0, 100) : jsonPayload,
                    e);
            return false;
        }
    }
}
